package com.navinfo.tripanalysis.offline.service.impl;

import com.navinfo.tripanalysis.common.vehicle.VehicleInfo;
import com.navinfo.tripanalysis.offline.service.LoadVehicleInfoService;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Properties;

/**
 * 通用加载车辆信息服务的抽象类
 */
public abstract class AbstractLoadVehicleInfoService implements LoadVehicleInfoService {
    protected Logger logger = LoggerFactory.getLogger(this.getClass());
    private String driver;
    private String user;
    private String pass;
    private String url;

    public String getDriver() {
        return driver;
    }

    public void setDriver(String driver) {
        this.driver = driver;
    }

    public String getUser() {
        return user;
    }

    public void setUser(String user) {
        this.user = user;
    }

    public String getPass() {
        return pass;
    }

    public void setPass(String pass) {
        this.pass = pass;
    }

    public String getUrl() {
        return url;
    }

    public void setUrl(String url) {
        this.url = url;
    }

    @Override
    public JavaPairRDD<Long, VehicleInfo> load(SparkSession session) {
        Properties property = new Properties();
        property.put("driver", getDriver());
        property.put("user", getUser());
        property.put("password", getPass());

        return getVehicleInfoPair(session, session.read().jdbc(getUrl(), getTable(), property).toJavaRDD());
    }

    /**
     * 查询语句
     * @return
     */
    protected abstract String getTable();


    /**
     * 查询结果转换为JavaPairRDD<Long, VehicleInfo>
     * @return
     */
    protected abstract JavaPairRDD<Long, VehicleInfo> getVehicleInfoPair(SparkSession session, JavaRDD<Row> rowJavaRDD);
}
